package avicit.bdp.dds.api.service;

import avicit.bdp.common.base.BaseService;
import avicit.bdp.common.dto.request.dds.ProcessSchedulerDto;
import avicit.bdp.common.utils.BdpLogUtil;
import avicit.bdp.common.utils.DateUtils;
import avicit.bdp.dds.api.dto.ScheduleParam;
import avicit.bdp.dds.api.enums.Status;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.dao.entity.ProcessDefinition;
import avicit.bdp.dds.dao.entity.Project;
import avicit.bdp.dds.dao.entity.ProjectUser;
import avicit.bdp.dds.dao.entity.Schedule;
import avicit.bdp.dds.dao.mapper.ProcessDefinitionMapper;
import avicit.bdp.dds.dao.mapper.ProjectMapper;
import avicit.bdp.dds.dao.mapper.ScheduleMapper;
import avicit.bdp.dds.dispatch.enums.FailureStrategy;
import avicit.bdp.dds.dispatch.enums.Priority;
import avicit.bdp.dds.dispatch.enums.ReleaseState;
import avicit.bdp.dds.dispatch.enums.WarningType;
import avicit.bdp.dds.dispatch.model.Server;
import avicit.bdp.dds.service.process.ProcessService;
import avicit.bdp.dds.service.quartz.ProcessScheduleJob;
import avicit.bdp.dds.service.quartz.QuartzExecutors;
import avicit.bdp.dds.service.quartz.cron.CronUtils;
import avicit.platform6.commons.utils.ComUtil;
import avicit.platform6.core.exception.BusinessException;
import avicit.platform6.core.rest.msg.QueryRespBean;
import com.github.pagehelper.Page;
import com.github.pagehelper.PageHelper;
import org.apache.commons.lang3.StringUtils;
import org.quartz.CronExpression;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.text.MessageFormat;
import java.text.ParseException;
import java.util.*;

/**
 * @author admin
 */
@Service
public class SchedulerService extends BaseService<ScheduleMapper, Schedule> {

    private static final Logger logger = LoggerFactory.getLogger(SchedulerService.class);

    private final ProcessService processService;
    private final ProcessDefinitionMapper processDefineMapper;
    private final MonitorService monitorService;
    private final ProjectUserService projectUserService;

    private final ProjectMapper projectMapper;

    @Autowired
    public SchedulerService(ProcessService processService,
                            ProcessDefinitionMapper processDefineMapper,
                            MonitorService monitorService,
                            ProjectUserService projectUserService,
                            ProjectMapper projectMapper) {

        this.processService = processService;
        this.processDefineMapper = processDefineMapper;
        this.monitorService = monitorService;
        this.projectUserService = projectUserService;
        this.projectMapper = projectMapper;
    }

    /**
     * 保存定时配置
     *
     * @param dto dto
     */
    @Transactional(rollbackFor = Exception.class)
    public boolean insertSchedule(ProcessSchedulerDto dto) {

        String processDefineId = dto.getProcessDefinitionId();

        // step1:检查流程定义是否存在且是否上线
        ProcessDefinition processDefinition = processDefineMapper.selectByPrimaryKey(processDefineId);
        if (processDefinition == null) {
            logger.error("the process definition is invalid, id={}", processDefineId);
            throw new BusinessException(String.format("the process definition is exist, id=%s", processDefineId));
        }

        // step2:定时参数合法性检查
        if (dto.getStartTime() == null || dto.getEndTime() == null) {
            logger.error("The schedule info is null");
            throw new BusinessException("the schedule info is null");
        }

        checkCronTime(dto.getStartTime(), dto.getEndTime(), dto.getCrontab());

        // step3:定时任务持久化
        Schedule scheduleObj = new Schedule(dto);

        insert(scheduleObj);

        // 记录日志
        BdpLogUtil.log4Insert(scheduleObj);

        return true;
    }

    private void checkCronTime(Date startTime, Date endTime, String cron) {
        if (StringUtils.isBlank(cron) || !CronExpression.isValidExpression(cron)) {
            logger.error("cron: " + cron + " verify failure");
            throw new BusinessException(cron + " verify failure");
        }

        if (DateUtils.differSec(startTime, endTime) == 0) {
            logger.error("The start time={} must not be the same as the end={}", startTime, endTime);
            throw new BusinessException(String.format("the start time is same with the end, start=%s, end=%s",
                    startTime, endTime));
        }
    }


    /**
     * 更新定时任务
     *
     * @param dto dto
     */
    @Transactional(rollbackFor = Exception.class)
    public boolean updateSchedule(ProcessSchedulerDto dto) {

        // step1:检查定时任务是否存在
        Schedule oldSchedule = selectByPrimaryKey(dto.getId());
        if (oldSchedule == null) {
            logger.error("scheduler {} does not exist", dto.getId());
            throw new BusinessException("scheduler " + dto.getId() + " does not exist");
        }

        String processDefinitionId = dto.getProcessDefinitionId();

        // step2:检查流程定义是否存在
        ProcessDefinition processDefinition = processDefineMapper.selectByPrimaryKey(processDefinitionId);
        if (processDefinition == null) {
            logger.error("the process definition is not exists, id:{}", processDefinitionId);
            throw new BusinessException(String.format("the process definition is not exists, id=%s", processDefinitionId));
        }

        // step3:检查定时时间
        checkCronTime(dto.getStartTime(), dto.getEndTime(), dto.getCrontab());

        Schedule scheduleObj = new Schedule(dto);
        // step4:持久化
        updateByPrimaryKey(getScheduleDto(scheduleObj,oldSchedule));

        //step5: 记录日志
        BdpLogUtil.log4Update(scheduleObj, oldSchedule);

        //发布的任务重新设置调度
        if (processDefinition.getReleaseState() == ReleaseState.ONLINE.getCode()) {
            deleteSchedule(processDefinition.getProjectId(), scheduleObj.getId());
            setSchedule(processDefinition.getProjectId(), scheduleObj);
        }

        return true;
    }

    private Schedule getScheduleDto(Schedule scheduleObj, Schedule oldSchedule) {
        oldSchedule.setStartTime(scheduleObj.getStartTime());
        oldSchedule.setEndTime(scheduleObj.getEndTime());
        oldSchedule.setCrontab(scheduleObj.getCrontab());
        oldSchedule.setWarningUserIds(scheduleObj.getWarningUserIds());
        oldSchedule.setWarningType(scheduleObj.getWarningType() == null ? WarningType.NONE.getCode() : scheduleObj.getWarningType());
        oldSchedule.setFailureStrategy(scheduleObj.getFailureStrategy() == null ? FailureStrategy.CONTINUE.getCode() : scheduleObj.getFailureStrategy());
        oldSchedule.setProcessInstancePriority(scheduleObj.getProcessInstancePriority() == null ? Priority.MEDIUM.getCode() : scheduleObj.getProcessInstancePriority());
        oldSchedule.setIsQuartzCron(scheduleObj.getIsQuartzCron());
        return oldSchedule;
    }

    /**
     * 查询定时任务
     *
     * @param processDefineId 流程定义ID
     * @param searchVal       关键词
     * @param pageNo          分页
     * @param pageSize        分页大小
     * @return 定时任务分页列表
     */
    public QueryRespBean<Schedule> querySchedule(String processDefineId, String searchVal,
                                                 Integer pageNo, Integer pageSize) {

        // step1:检查流程定义是否存在
        validProcessDefinitionExists(processDefineId);

        PageHelper.startPage(pageNo, pageSize);
        QueryRespBean<Schedule> queryRespBean = new QueryRespBean<>();
        Page<Schedule> schedulePage = this.mapper.queryByProcessDefineIdPaging(processDefineId, searchVal);
        queryRespBean.setResult(schedulePage);
        BdpLogUtil.log4Query(queryRespBean);
        return queryRespBean;
    }

    private void validProcessDefinitionExists(String processDefineId) {
        ProcessDefinition processDefinition = processDefineMapper.selectByPrimaryKey(processDefineId);
        if (processDefinition == null) {
            logger.error("the process definition is not exists, id:{}", processDefineId);
            throw new BusinessException(String.format("the process definition is not exists, id=%s", processDefineId));
        }
    }


    /**
     * set schedule online or offline
     *
     * @param projectName    project name
     * @param id             scheduler id
     * @param scheduleStatus schedule status
     * @return publish result code
     */
    @Transactional(rollbackFor = RuntimeException.class)
    public Map<String, Object> setScheduleState(String projectName, String id, ReleaseState scheduleStatus) {

        Map<String, Object> result = new HashMap<>(5);

        // check schedule exists
        Schedule scheduleObj = this.mapper.selectByPrimaryKey(id);

        if (scheduleObj == null) {
            throw new BusinessException(MessageFormat.format("调度配置定时表达式[{0}]不存在", id));
        }

        ProcessDefinition processDefinition = processDefineMapper.selectByPrimaryKey(scheduleObj.getProcessDefinitionId());
        if (processDefinition == null) {
            throw new BusinessException(MessageFormat.format("工作流定义[{0}]不存在", scheduleObj.getProcessDefinitionId()));
        }

        if (scheduleStatus == ReleaseState.ONLINE) {
            // check process definition release state
            if (ReleaseState.of(processDefinition.getReleaseState()) != ReleaseState.ONLINE) {
                logger.info("not release process definition id: {} , name : {}", processDefinition.getId(), processDefinition.getName());
                throw new BusinessException(MessageFormat.format("工作流定义[{0}]不是上线状态", processDefinition.getName()));
            }
            // check sub process definition release state
            List<String> subProcessDefineIds = new ArrayList<>();
            processService.recurseFindSubProcessId(scheduleObj.getProcessDefinitionId(), subProcessDefineIds);
            String[] idArray = subProcessDefineIds.toArray(new String[0]);
            if (subProcessDefineIds.size() > 0) {
                List<ProcessDefinition> subProcessDefinitionList = processDefineMapper.queryDefinitionListByIdList(idArray);
                if (subProcessDefinitionList != null && subProcessDefinitionList.size() > 0) {
                    for (ProcessDefinition subProcessDefinition : subProcessDefinitionList) {
                        //if there is no online process, exit directly
                        if (ReleaseState.of(subProcessDefinition.getReleaseState()) != ReleaseState.ONLINE) {
                            logger.info("not release process definition id: {} , name : {}",
                                    subProcessDefinition.getId(), subProcessDefinition.getName());
                            throw new BusinessException(MessageFormat.format("工作流定义[{0}]不是上线状态",
                                    subProcessDefinition.getId()));
                        }
                    }
                }
            }
        }

        // check master server exists
        List<Server> masterServers = monitorService.getServerListFromZK(true);

        if (masterServers.size() == 0) {
            throw new BusinessException(Status.MASTER_NOT_EXISTS.getMsg());
        }

        try {
            switch (scheduleStatus) {
                case ONLINE: {
                    logger.info("Call master client set schedule online, project id: {}, flow id: {},host: {}",
                            projectName, processDefinition.getId(), masterServers);
                    setSchedule(projectName, scheduleObj);
                    break;
                }
                case OFFLINE: {
                    logger.info("Call master client set schedule offline, project id: {}, flow id: {}, host: {}",
                            projectName, processDefinition.getId(), masterServers);

                    deleteSchedule(projectName, id);
                    break;
                }
                default: {
                    throw new BusinessException(MessageFormat.format(Status.SCHEDULE_STATUS_UNKNOWN.getMsg(), scheduleStatus.toString()));
                }
            }
        } catch (Exception e) {
            result.put(Constants.MSG, scheduleStatus == ReleaseState.ONLINE ? "set online failure" : "set offline failure");
            throw new BusinessException(result.get(Constants.MSG).toString());
        }

        return result;
    }

    /**
     * 删除定时任务
     *
     * @param projectId  项目空间ID
     * @param scheduleId 定时任务ID
     * @throws RuntimeException 异常
     */
    private void deleteSchedule(String projectId, String scheduleId) throws RuntimeException {
        logger.info("delete schedules of project id:{}, schedule id:{}", projectId, scheduleId);

        String jobName = QuartzExecutors.buildJobName(scheduleId);
        String jobGroupName = QuartzExecutors.buildJobGroupName(projectId);

        if (!QuartzExecutors.getInstance().deleteJob(jobName, jobGroupName)) {
            logger.warn("set offline failure:projectId:{},scheduleId:{}", projectId, scheduleId);
            throw new RuntimeException("set offline failure");
        }
    }

    private void setSchedule(String projectId, Schedule schedule) {

        String scheduleId = schedule.getId();
        logger.info("set schedule, project id: {}, scheduleId: {}", projectId, scheduleId);

        Date startDate = schedule.getStartTime();
        Date endDate = schedule.getEndTime();

        String jobName = QuartzExecutors.buildJobName(scheduleId);
        String jobGroupName = QuartzExecutors.buildJobGroupName(projectId);

        Map<String, Object> dataMap = QuartzExecutors.buildDataMap(projectId, scheduleId, schedule);

        QuartzExecutors.getInstance().addJob(ProcessScheduleJob.class, jobName, jobGroupName, startDate, endDate,
                schedule.getCrontab(), dataMap);
    }

    /**
     * 根据ID删除定时任务
     *
     * @param scheduleId 定时任务ID
     * @return result
     */
    public boolean deleteScheduleById(String scheduleId) {

        Schedule schedule = selectByPrimaryKey(scheduleId);
        if (schedule == null) {
            logger.error("scheduler cron tab {} does not exist", scheduleId);
            throw new BusinessException(String.format("scheduler cron=%s doesn't exist.", scheduleId));
        }

        deleteByPrimaryKey(scheduleId);

        BdpLogUtil.log4Delete(schedule);

        return true;
    }

    /**
     * 预览调度配置
     *
     * @param scheduleParam scheduleParam
     * @return date list
     */
    public List<Date> previewSchedule(ScheduleParam scheduleParam) {
        CronExpression cronExpression;
        Date now = new Date();

        Date startTime = now.after(scheduleParam.getStartTime()) ? now : scheduleParam.getStartTime();
        Date endTime = scheduleParam.getEndTime();
        try {
            cronExpression = CronUtils.parse2CronExpression(scheduleParam.getCrontab());
        } catch (ParseException e) {
            logger.error("解析调度表达式错误");
            throw new BusinessException("解析调度表达式错误");
        }

        return CronUtils.getSelfFireDateList(startTime, endTime, cronExpression,
                Constants.PREVIEW_SCHEDULE_EXECUTE_COUNT);
    }

    /**
     * 根据流程定义ID查询定时任务
     *
     * @param processDefinitionId 流程定义ID
     * @return 流程定义下所有的定时任务
     */
    public List<Schedule> queryByProcessDefinitionId(String processDefinitionId) {
        return this.mapper.queryByProcessDefinitionId(processDefinitionId);
    }

    /**
     * 获取项目用户列表
     *
     * @param projectId 项目空间ID
     * @return 项目用户列表
     */
    public List<ProjectUser> getProjectUserList(String projectId) {
        ProjectUser userQuery = new ProjectUser();
        userQuery.setProjectId(projectId);
        List<ProjectUser> userList = projectUserService.selectList(userQuery);
        //增加项目创建人
        Project project = projectMapper.selectByPrimaryKey(projectId);
        if (project != null && StringUtils.isNotBlank(project.getCreatedBy())) {
            ProjectUser user = new ProjectUser();
            user.setId(ComUtil.getId());
            user.setUserId(project.getCreatedBy());
            user.setProjectId(projectId);
            userList.add(user);
        }

        return userList;
    }

}
